package kafkastudy.spring;


import org.junit.Before;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import java.util.concurrent.ExecutionException;

/**
 * Created by hp on 2017/9/30.
 */
public class KafkaTest {

    ApplicationContext applicationContext;
    KafkaTemplate template;

    @Before
    public void before() {
        applicationContext = new ClassPathXmlApplicationContext("spring/kafka-producer.xml", "spring/kafka-consumer.xml");
        template = (KafkaTemplate) applicationContext.getBean("KafkaTemplate");

    }

    @Test
    public void test1() {
        ListenableFuture<SendResult<String, String>> listenableFuture = template.send("xmine", "this is test");
        try {
            SendResult<String, String> str = listenableFuture.get();
            //发送成功回调
            SuccessCallback<SendResult<String, String>> successCallback = new SuccessCallback<SendResult<String, String>>() {
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    //成功业务逻辑
                    System.out.println("发送成功回调");
                }
            };
            //发送失败回调
            FailureCallback failureCallback = new FailureCallback() {
                @Override
                public void onFailure(Throwable ex) {
                    //失败业务逻辑
                    System.out.println("发送失败回调");
                }
            };
            listenableFuture.addCallback(successCallback, failureCallback);
            System.out.println(str.getProducerRecord().key());
            System.out.println(str.getProducerRecord().value());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        while (true) {

        }
    }

    @Test
    public void test2() {
        while (true) {

        }
    }
}
